package com.upcoming.data.storm.fileRead;

import com.upcoming.data.storm.util.MyUtil;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-23 10:06
 * 类描述：
 * 修改人：
 * 修改时间：
 */
public class FileBolt implements IRichBolt {

    private OutputCollector outputCollector;


    public FileBolt() {
        MyUtil.outLog2NC(this, "new FileBolt()");
    }


    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        MyUtil.outLog2NC(this, "prepare()");
        this.outputCollector = outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {
        String line = tuple.getString(0);
        MyUtil.outLog2NC(this, line);
        if (line != null) {
            String[] words = line.split(" ");
            for (String word : words) {
                try {
                    this.outputCollector.emit(new Values(word));
                } catch (Exception e) {
                    MyUtil.outLog2NC(this, e.getMessage());
                }
            }
        }
    }

    @Override
    public void cleanup() {

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("word"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
